﻿#include "EventCache.h"

#include "event2/util.h"
#include "event2/event_compat.h"
#include "event2/event_struct.h"
#include "event2/bufferevent.h"
#include <event2/buffer.h>
#include <event2/listener.h>
#include <fstream>

#ifndef WIN32 
#include <string.h>
#include <sys/socket.h>
#include <netinet/in.h>
#include <arpa/inet.h>
#include <unistd.h>
#include <sys/types.h>
#include <sys/socket.h>
#include <fcntl.h>
#endif

#include <stdexcept>

EventCache::EventCache(int size)
{
    if (size <= 0) throw std::runtime_error("bad size");

    m_data = new char[size];
    m_send_data = new char[size];
    m_socket = -1;
    m_size = size;
    m_start = 0;
    m_end = 0;
    m_start_send = 0;
    m_end_send = 0;

    m_status = 0;
    m_general_evwent = nullptr; 
    m_write_event = nullptr;

    m_readcb = nullptr;
    m_eventcb = nullptr;
    m_user_data = nullptr;
}

EventCache::~EventCache()
{
    shutdown();
}

int EventCache::init(struct event_base* base, int ip, int port)
{
    if (base == nullptr || m_status) return -1;

    m_socket = ConnectServer(ip, port);
    if (m_socket == -1)
    {
        return -1;
    }
    m_base = base;

    struct timeval tm { 0, 500000 }; /* 500ms */
    m_write_event = event_new(
        base,
        m_socket,
        EV_WRITE | EV_TIMEOUT | EV_PERSIST,
        [](evutil_socket_t fd, short events, void* arg)
        {
            EventCache* cache = reinterpret_cast<EventCache*>(arg);
            if(events & EV_TIMEOUT) return cache->ConnectEvent(fd, CACHE_CONNECT_TIMEOUT);
            else
                return cache->ConnectEvent(fd, CACHE_CONNECT);
        },
        reinterpret_cast<void*>(this)
            );

    event_add(m_write_event, &tm);
    return 0;
}

int EventCache::init(struct event_base* base, int nsocket)
{
    if ((base == nullptr && m_base == nullptr) || nsocket == -1 || m_status) return -1;

    m_socket = nsocket;
    m_base = base;
    
    struct timeval tm { 0, 1000 };
    m_general_evwent = event_new(
        m_base,
        m_socket,
        EV_READ | EV_CLOSED | EV_PERSIST | EV_TIMEOUT,
        [](evutil_socket_t fd, short events, void* arg)
        {
            reinterpret_cast<EventCache*>(arg)->EventProssess(fd, events);
        },
        reinterpret_cast<void*>(this)
            );
    
    if (m_general_evwent == NULL)
    {
        return -1;
    }
    m_status = 1;
    event_add(m_general_evwent, &tm);

    return 0;
}

void EventCache::SetCallBack(
    std::function<void(void*)> readcb, std::function<void(int, void*)> eventcb, void* user_data
)
{
    m_readcb = readcb;
    m_eventcb = eventcb;
    m_user_data = user_data;
}

void EventCache::shutdown()
{
    if (-1 != m_socket && 0 != m_status)
    {
#ifdef WIN32
        closesocket(m_socket);
#else
        close(m_socket);
#endif
        m_socket = -1;
    }

    if (m_general_evwent)
    {
        event_del(m_general_evwent);
        event_free(m_general_evwent);
        m_general_evwent = nullptr;
    }

    if (m_data) delete[] m_data;
    if (m_send_data) delete[] m_send_data;

    m_status = 0;
    m_data = nullptr;
    m_send_data = nullptr;
}

int EventCache::ConnectServer(int ip, int port)
{
    int mode = 1;
    struct sockaddr_in sin = { 0 };
    
    int sockClient = socket(AF_INET, SOCK_STREAM, 0);
#ifdef WIN32
    if (0 != ioctlsocket(sockClient, FIONBIO, reinterpret_cast<u_long FAR*>(&mode)))
    {
        closesocket(sockClient);
        return -1;
    }
#else
    mode = fcntl(sockClient, F_GETFL, 0);
    if (0 != fcntl(sockClient, F_SETFL, O_NONBLOCK | mode))
    {
        close(socket);
        return -1;
    }
#endif

    sin.sin_family = AF_INET;
    sin.sin_port = htons(port);
    sin.sin_addr.s_addr = htonl(ip);

    if (-1 != connect(sockClient, (sockaddr*)&sin, sizeof(sockaddr)))
    {
        return sockClient;
    }
    return sockClient;
}

void EventCache::ConnectEvent(int nSokcet, int events)
{
    do 
    {
        if (events == CACHE_CONNECT_TIMEOUT)
        {
            m_eventcb(CACHE_CONNECT_TIMEOUT, m_user_data);
            break;
        }

        if (0 != init(m_base, nSokcet))
        {
            m_eventcb(CACHE_CONNECT_ERROR, m_user_data);
        }
        else
        {
            m_status = 1;
            m_eventcb(CACHE_CONNECT, m_user_data);
        }
    } while (0);
    
    if (m_write_event)
    {
        event_del(m_write_event);
        event_free(m_write_event);
        m_write_event = nullptr;
    }
}

void EventCache::EventProssess(int nSocket, short events)
{
    int type = CACHE_PROBE;
    do 
    {
        if (events & EV_READ)
        {
            if (RWriteCache(nSocket) < 0)
            {
                type = CACHE_READ_ERROR;
                break;
            }
            else
            {
                return m_readcb(m_user_data);
            }
        }
        
        if (events & EV_TIMEOUT)
        {
            if (SReadCache(nSocket) < 0)
            {
                type = CACHR_WRITE_ERROR;
                break;
            }
            else
                return;
        }

        if (events & EV_CLOSED) type = CACHE_CLOSE_RES;

    } while (0);

    event_del(m_general_evwent);
    m_eventcb(type, m_user_data);
    
#ifdef WIN32
    closesocket(m_socket);
#else
    close(m_socket);
#endif
    m_socket = -1;
}

int EventCache::SendData(const char* buff, int bufflen)
{
    if (!m_status) return -1;

    if (!IsEmptyCache(m_start_send, m_end_send))
    {
        return SWriteCache(buff, bufflen);
    }

    int realsend = send(m_socket, buff, bufflen, 0);
    if (-1 == realsend)
    {
        return -1;
    }
    if (realsend < bufflen)
    {
        realsend += SWriteCache(buff + realsend, bufflen - realsend);
    }

    return realsend;
}

int EventCache::ReadData(char* buff, int buffsize)
{
    if (!m_status) return -1;

    return RReadCache(buff, buffsize);
}

int EventCache::RReadCache(char* buff, int buffsize)
{
    if (m_start == m_end) return 0;

    int start = m_start.load(), end = m_end.load();
    int curdata = 0;

    if (start > end)
        curdata = m_size - start + end;
    else
        curdata = end - start;

    curdata = curdata > buffsize ? buffsize : curdata;

    if (start > end)
    {
        if (m_size - start < curdata)
        {
            memcpy(buff, m_data + start, m_size - start);
            memcpy(buff + m_size - start, m_data, curdata - m_size + start);
        }
        else
            memcpy(buff, m_data + start, curdata);
    }
    else
    {
        memcpy(buff, m_data + start, curdata);
    }

    start = (start + curdata) % m_size;
    m_start = start;
    end = m_end.load();

    return curdata; /* 实际读取的数据长度  */
}

int EventCache::RWriteCache(int nSocket)
{
    int start = m_start.load(), end = m_end.load();
    int curcal = 0;

    if (start == (end + 1) % m_size) return 0;

    if (start > end)
        curcal = start - end - 1;
    else
        curcal = m_size - end + start - 1;

    std::shared_ptr<char> tempbuf = std::shared_ptr<char>(
        new char[curcal + 1], [](char* p) {delete[] p; }
    );

    curcal = recv(nSocket, tempbuf.get(), curcal, 0);
    if (curcal <= 0)
    {
        return -1;
    }

    if (start > end) 
    {
        memcpy(m_data + end, tempbuf.get(), curcal);
    }
    else
    {
        if (m_size - end > curcal)
        {
            memcpy(m_data + end, tempbuf.get(), curcal);
        }
        else
        {
            memcpy(m_data + end, tempbuf.get(), m_size - end);
            memcpy(m_data, tempbuf.get() + m_size - end, curcal - m_size + end);
        }
    }

    end = (end + curcal) % m_size;
    m_end = end;
    start = m_start.load();

    return curcal;
}

int EventCache::SReadCache(int nSocket)
{
    if (m_start_send == m_end_send) return 0;

    int start = m_start_send.load(), end = m_end_send.load();
    int realsize = 0, ret;

    do 
    {
        if (start > end)
        {
            ret = send(nSocket, m_send_data + start, m_size - start, 0);
            if (-1 == ret) break;

            realsize += ret;
            if (realsize != m_size - start) break;

            ret = send(nSocket, m_send_data + start, m_size - start, 0);
            if (-1 == ret) break;
            realsize += ret;
        }
        else
        {
            ret = send(nSocket, m_send_data + start, end - start, 0);
            if (-1 == ret)
            {
                break;
            }
            realsize += ret;
        }
    } while (0);
    
    start = (start + realsize) % m_size;
    m_start_send = start;
    end = m_end_send.load();

    if (ret == -1) return -1;

    return m_end_send != m_start_send; /* 是否有剩余数据  */
}

int EventCache::SWriteCache(const char* buff, int buffsize)
{
    int start = m_start_send.load(), end = m_end_send.load();
    int curcal = 0;

    if (start == (end + 1) % m_size) return 0; /* 缓冲区满 */

    if (start > end)
        curcal = start - end - 1;
    else
        curcal = m_size - end + start - 1;
    curcal = buffsize > curcal ? curcal : buffsize;

    if (start > end)
    {
        memcpy(m_send_data + end, buff, curcal);
    }
    else
    {
        if (m_size - end > curcal)
        {
            memcpy(m_send_data + end, buff, curcal);
        }
        else
        {
            memcpy(m_send_data + end, buff, m_size - end);
            memcpy(m_send_data, buff + m_size - end, curcal - m_size + end);
        }
    }

    end = (end + curcal) % m_size;
    m_end_send = end;
    start = m_start_send.load();

    return curcal;   /* 实际写入到缓冲区的数据 */
}